package com.hex.ds.hdtp.core.inf.metadata.service.impl;

import com.hex.ds.hdtp.core.inf.common.enums.FieldType;
import com.hex.ds.hdtp.core.inf.common.utils.JdbcConnectUtil;
import com.hex.ds.hdtp.core.inf.common.utils.JdbcUtil;
import com.hex.ds.hdtp.core.inf.metadata.converter.IImpalaSqlResultMapper;
import com.hex.ds.hdtp.core.inf.metadata.entity.TableFieldsEntity;
import com.hex.ds.hdtp.core.inf.metadata.po.SourceConfInfoPo;
import com.hex.ds.hdtp.core.inf.metadata.repository.ISourceConfInfoRepository;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @Package com.hex.ds.hdtp.core.inf.metadata.service.impl
 * @ClassName ImpalaMethodService
 * @Description Impala元数据管理服务类
 * @Author gj.xu
 * @Date 2024/3/4 18:20
 * @Version v2.0
 **/
@Slf4j
@Service("IMPALA")
public class ImpalaMethodService extends AbstractHiveQLMethodService {

    private final IImpalaSqlResultMapper impalaSqlResultMapper;

    public ImpalaMethodService(ISourceConfInfoRepository sourceConfInfoRepository, IImpalaSqlResultMapper impalaSqlResultMapper) {
        super(sourceConfInfoRepository);
        this.impalaSqlResultMapper = impalaSqlResultMapper;
    }

    @Override
    protected TableFieldsEntity toTableFieldsBean(Map<String, String> tableFieldsMap) {
        return impalaSqlResultMapper.toTableFieldsBean(tableFieldsMap);
    }

    @Override
    public List<TableFieldsEntity> tableStructure(String sourceCode, String schemaName, String tableName) {
        List<SourceConfInfoPo> sourceConfInfoPoList = getSourceConfInfoRepository().queryListBySourceCode(sourceCode);
        List<TableFieldsEntity> resultList = new ArrayList<>();
        String querySql = queryTableStructureSql(schemaName, tableName);
        try (Connection jdbcConnection = JdbcUtil.getJdbcConnection(sourceConfInfoPoList);
             Statement jdbcStatement = JdbcUtil.createJdbcStatement(jdbcConnection);
             ResultSet resultSet = JdbcUtil.executeQuerySql(jdbcStatement, querySql)
        ) {
            List<String> partitionColumns = getPartitionColumns(schemaName, tableName, sourceCode);
            resultList = processResultSet(resultSet, resultList, partitionColumns);
        } catch (SQLException e) {
            log.error("获取表结构失败", e.getStackTrace());
        }
        return resultList;
    }

    public List<TableFieldsEntity> tableStructureByType(String sourceCode, String schemaName, String tableName, String fieldType) {
        List<SourceConfInfoPo> sourceConfInfoPoList = getSourceConfInfoRepository().queryListBySourceCode(sourceCode);
        List<TableFieldsEntity> resultList = new ArrayList<>();
        String querySql = queryTableStructureSql(schemaName, tableName);
        try (Connection jdbcConnection = JdbcUtil.getJdbcConnection(sourceConfInfoPoList);
             Statement jdbcStatement = JdbcUtil.createJdbcStatement(jdbcConnection);
             ResultSet resultSet = JdbcUtil.executeQuerySql(jdbcStatement, querySql)
        ) {
            List<String> partitionColumns = getPartitionColumns(schemaName, tableName, sourceCode);
            resultList = processResultSet(resultSet, resultList, partitionColumns);
        } catch (SQLException e) {
            log.error("根据类型获取表结构失败", e.getStackTrace());
        }
        List<String> fieldTypes = FieldType.fromFieldType(fieldType);
        List<TableFieldsEntity> afterDealFields = resultList.stream().filter(e -> fieldTypes.contains(e.getColumnType().toUpperCase())).collect(Collectors.toList());
        return afterDealFields;
    }

    /**
     * @Method processResultSet <br>
     * @Param resultSet SQL查询的ResultSet对象
     * @Param resultList 要用字段信息填充的列表
     * @Return List<TableFieldsEntity>  <br>
     * @Description 将tableFieldsMap放入resultList中，如果出现null字段则停止<br>
     * @Author Yin.Yang<br>
     * @Date 2023/9/6 16:46 <br>
     * @Version V1.0<br>
     */
    private List<TableFieldsEntity> processResultSet(ResultSet resultSet, List<TableFieldsEntity> resultList, List<String> partitionColumns) throws SQLException {

        while (resultSet.next()) {
            Map<String, String> tableFieldsMap = new HashMap<>();
            if (extractTableFields(resultSet, tableFieldsMap, partitionColumns, "type")) {
                resultList.add(toTableFieldsBean(tableFieldsMap));
            }
        }
        return resultList;
    }

    /**
     * @param schemaName 数据源中的库名
     * @param tableName  数据源中的表名
     * @Method getPartitionColumns <br>
     * @Return 分区字段的Set集合<String> <br>
     * @Description 获取Impala分区字段<br>
     * @Author Yin.Yang<br>
     * @Date 2023/9/12 11:38 <br>
     * @Version V1.0<br>
     */
    public List<String> getPartitionColumns(String schemaName, String tableName, String sourceCode) throws SQLException {
        List<SourceConfInfoPo> sourceConfInfoPoList = getSourceConfInfoRepository().queryListBySourceCode(sourceCode);
        String querySql = this.getTablePartitionFieldSql(schemaName, tableName);
        List<String> allFieldsName = null;
        try {
            allFieldsName = JdbcConnectUtil.getFieldResultList(sourceConfInfoPoList, querySql);
            return allFieldsName.subList(0, allFieldsName.indexOf("#ROWS"));
        } catch (Exception e) {
            log.warn(e.getMessage());
            return allFieldsName;
        }
    }

}
